package com.atguigu.dga.governance.assessor.quality;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga.constant.DgaConstant;
import com.atguigu.dga.governance.assessor.Assessor;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.math.BigDecimal;
import java.net.URI;
import java.util.Date;
import java.util.List;

/**
 * @Description: 表产出数据量监控
 * 必须日分区表 当日产出的数据量，超过前{days}天平均产出量{upper_limit}% ，或低于{lower_limit}%  ，则给0分，其余10分
 * @Author: lay
 * @Date: 2024/6/29 0:03
 */
@Component("TABLE_PRODUCE_DATA")
public class TableProduceDataAssessor extends Assessor {
    @Override
    public void checkProblem(GovernanceAssessDetail governanceAssessDetail, AssessParam assessParam) throws Exception {
        //获取参数
        String paramsJson = assessParam.getGovernanceMetric().getMetricParamsJson();
        JSONObject paramJson = JSONObject.parseObject(paramsJson);
        Integer days = paramJson.getInteger("days");
        Integer upperLimit = paramJson.getInteger("upper_limit");
        Integer lowerLimit = paramJson.getInteger("lower_limit");
        
        //生命周期类型
        String lifecycleType = assessParam.getTableMetaInfo().getTableMetaInfoExtra().getLifecycleType();
        //非日分区直接返回
        if (!lifecycleType.equals(DgaConstant.LIFECYCLE_TYPE_DAY)){
            return;
        }
        //表路径
        String tableFsPath = assessParam.getTableMetaInfo().getTableFsPath();
        //考评日期
        Date assessDate = DateUtils.parseDate(assessParam.getAssessDate(), "yyyy-MM-dd");
        //最近分区日期
        Date dtDate = DateUtils.addDays(assessDate, -1);
        String dtDateString = DateFormatUtils.format(dtDate, "yyyy-MM-dd");
        
        //取表分区字段
        String partitionColNameJson = assessParam.getTableMetaInfo().getPartitionColNameJson();
        List<JSONObject> colNameList = JSON.parseArray(partitionColNameJson, JSONObject.class);
        //分区字段
        String partitionColName = colNameList.get(0).getString("name");
        
        //完整路径
        String fullPath = tableFsPath + "/" + partitionColName + "=" + dtDateString;
        FileSystem fs = FileSystem.get(new URI(fullPath), new Configuration(), assessParam.getTableMetaInfo().getTableFsOwner());
        FileStatus[] fileStatuses = fs.listStatus(new Path(fullPath));
        //当日产出数据量
        long dataSize = calcDataSize(fileStatuses, fs, 0L);

        long beforeTotalDataSize = 0L;
        int realBeforeDays = 0;
        for (Integer i = 1; i <= days; i++) {
            Date beforeDate = DateUtils.addDays(dtDate, -i);
            String beforeDateString = DateFormatUtils.format(beforeDate, "yyyy-MM-dd");
            //完整路径
            String beforeFullPath = tableFsPath + "/" + partitionColName + "=" + beforeDateString;
            
            //判断路径存不存在
            if (!fs.exists(new Path(beforeFullPath))){
                continue;
            }

            long beforeDataSize = calcDataSize(fs.listStatus(new Path(beforeFullPath)), fs, 0L);
            beforeTotalDataSize += beforeDataSize;
            //记录真实存在的之前的天
            realBeforeDays++;
        }
        
        if (realBeforeDays > 0){
            //平均产出量
            long avgDataSize = beforeTotalDataSize / realBeforeDays;
            if ((dataSize - avgDataSize) * 100 / avgDataSize > upperLimit){
                governanceAssessDetail.setAssessProblem("当日产出数据量超过前 " + days + "天的平均产出数据量的 " + upperLimit + "%");
            } else if ((dataSize - avgDataSize) * 100 / avgDataSize < lowerLimit){
                governanceAssessDetail.setAssessProblem("当日产出数据量低于前 " + days + "天的平均产出数据量的 " + lowerLimit + "%");
            }
            governanceAssessDetail.setAssessScore(BigDecimal.ZERO);
            governanceAssessDetail.setAssessComment("当前产出数据量: " + dataSize + " , 实际前 " + realBeforeDays + "天的平均产出数据量: " + avgDataSize);
        }

    }

    /**
     * 递归求dataSize
     * @param fileStatuses
     * @param fs
     * @param dataSize
     * @return
     * @throws IOException
     */
    private long calcDataSize(FileStatus[] fileStatuses, FileSystem fs, long dataSize) throws IOException {
        for (FileStatus fileStatus : fileStatuses) {
            if (fileStatus.isFile()){
                dataSize += fileStatus.getLen();
            } else {
                FileStatus[] subFileStatuses = fs.listStatus(fileStatus.getPath());
                return calcDataSize(subFileStatuses, fs, dataSize);
            }
        }
        return dataSize;
    }
}
